package club.bigtian.sync.service.impl;

import club.bigtian.sync.config.SyncConfig;
import club.bigtian.sync.model.ColumnInfo;
import club.bigtian.sync.model.DataRow;
import club.bigtian.sync.model.TableSchema;
import club.bigtian.sync.service.SyncService;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.event.ApplicationEnvironmentPreparedEvent;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.File;

import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.SQLException;

/**
 * 同步服务实现类
 * 实现同步服务接口
 */
@Slf4j
@Service
public class SyncServiceImpl implements SyncService,  ApplicationListener<ApplicationReadyEvent>  {

    @Autowired
    @Qualifier("mysqlJdbcTemplate")
    private JdbcTemplate mysqlJdbcTemplate;

    @Autowired
    @Qualifier("clickhouseJdbcTemplate")
    private JdbcTemplate clickhouseJdbcTemplate;

    @Autowired
    private SyncConfig syncConfig;

    /**
     * 表结构缓存
     */
    private final Map<String, TableSchema> tableSchemaCache = new ConcurrentHashMap<>();

    /**
     * binlog客户端
     */
    private BinaryLogClient binlogClient;

    /**
     * 是否正在运行增量同步
     */
    private volatile boolean isRunning = false;

    /**
     * 表ID到表名的映射
     */
    private final Map<Long, String> tableIdToNameMap = new HashMap<>();

    /**
     * 初始化同步环境
     */
    @Override
    public void init() {
        log.info("初始化同步环境");
        
        // 获取所有表的结构
        for (String tableName : syncConfig.getTableList()) {
            TableSchema tableSchema = getTableSchema(tableName);
            tableSchemaCache.put(tableName, tableSchema);
            
            // 创建ClickHouse表
            createClickHouseTable(tableSchema);
        }
        
        // 执行全量同步
        if (syncConfig.isFullImportEnabled() && syncConfig.isFullSync()) {
            for (String tableName : syncConfig.getTableList()) {
                fullSync(tableName);
            }
        }
        
        // 执行增量同步
        if (syncConfig.isIncrementalSync() && syncConfig.isBinlogEnabled()) {
            startIncrementalSync();
        }
    }

    /**
     * 执行全量同步
     * 
     * @param tableName 表名
     */
    @Override
    public void fullSync(String tableName) {
        log.info("开始全量同步表: {}", tableName);
        
        TableSchema tableSchema = tableSchemaCache.getOrDefault(tableName, getTableSchema(tableName));
        if (tableSchema == null) {
            log.error("表结构不存在: {}", tableName);
            return;
        }
        
        // 分批次查询数据
        int batchSize = syncConfig.getBatchSize();
        int offset = 0;
        
        while (true) {
            // 查询一批数据
            String sql = String.format("SELECT * FROM %s LIMIT %d, %d", tableName, offset, batchSize);
            List<DataRow> dataRows = mysqlJdbcTemplate.query(sql, (rs, rowNum) -> {
                DataRow dataRow = new DataRow();
                dataRow.setTableName(tableName);
                
                for (ColumnInfo columnInfo : tableSchema.getColumns()) {
                    String columnName = columnInfo.getName();
                    Object value = rs.getObject(columnName);
                    dataRow.setValue(columnName, value);
                }
                
                // 设置 version 为当前时间戳，isDeleted 为 0
                dataRow.setVersion(System.currentTimeMillis());
                dataRow.setIsDeleted((byte) 0);
                
                return dataRow;
            });
            
            // 如果没有数据，跳出循环
            if (dataRows.isEmpty()) {
                break;
            }
            
            // 插入ClickHouse
            batchInsertToClickHouse(tableSchema, dataRows);
            
            // 更新偏移量
            offset += dataRows.size();
            log.info("同步表 {} 进度: {} 条", tableName, offset);
            
            // 如果批次数据量小于批次大小，跳出循环
            if (dataRows.size() < batchSize) {
                break;
            }
        }
        
        log.info("完成全量同步表: {}, 共同步 {} 条数据", tableName, offset);
    }

    /**
     * 启动增量同步
     */
    @Override
    public void startIncrementalSync() {
        if (isRunning) {
            log.warn("增量同步已经在运行中");
            return;
        }

        String host = syncConfig.getMysqlHost();
        int port = syncConfig.getMysqlPort();
        String username = syncConfig.getMysqlUsername();
        String password = syncConfig.getMysqlPassword();
        String database = syncConfig.getMysqlDatabase();

        try {
            // 初始化表ID到表名的映射
            initTableIdMap(host, port, username, password, database);

            // 创建binlog客户端
            binlogClient = new BinaryLogClient(host, port, username, password);

            // 获取当前MySQL的binlog状态
            String[] currentBinlogStatus = getBinlogStatus();
            if (currentBinlogStatus == null) {
                log.error("无法获取当前MySQL的binlog状态，请确保MySQL已开启binlog");
                return;
            }

            String currentBinlogFile = currentBinlogStatus[0];
            long currentBinlogPos = Long.parseLong(currentBinlogStatus[1]);

            // 获取配置中的binlog文件名和位置
            String configBinlogFile = syncConfig.getBinlogFilename();
            long configBinlogPos = syncConfig.getBinlogPosition();

            // 获取上次同步的binlog文件名和位置
            String lastBinlogFile = syncConfig.getLastBinlogFilename();
            long lastBinlogPos = syncConfig.getLastBinlogPosition();

            log.info("当前MySQL binlog位置: {}:{}", currentBinlogFile, currentBinlogPos);
            log.info("配置的binlog位置: {}:{}", configBinlogFile, configBinlogPos);
            log.info("上次同步的binlog位置: {}:{}", lastBinlogFile, lastBinlogPos);

            // 决定使用哪个binlog位置
            String useBinlogFile;
            long useBinlogPos;

            // 如果配置中指定了binlog文件和位置
            if (StringUtils.hasText(configBinlogFile) && configBinlogPos > 0) {
                // 检查配置的binlog文件是否与当前的相同
                if (configBinlogFile.equals(currentBinlogFile)) {
                    // 如果相同，使用配置的position
                    useBinlogFile = configBinlogFile;
                    useBinlogPos = configBinlogPos;
                    log.info("使用配置的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                } else {
                    // 如果不同，检查是否有上次同步的位置
                    if (StringUtils.hasText(lastBinlogFile) && lastBinlogPos > 0) {
                        // 如果有上次同步的位置，检查上次同步的binlog文件是否与当前的相同
                        if (lastBinlogFile.equals(currentBinlogFile)) {
                            // 如果相同，使用上次同步的position
                            useBinlogFile = lastBinlogFile;
                            useBinlogPos = lastBinlogPos;
                            log.info("配置的binlog文件已变化，使用上次同步的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                        } else {
                            // 如果不同，说明binlog文件已经发生了多次轮换，使用当前的binlog位置
                            useBinlogFile = currentBinlogFile;
                            useBinlogPos = currentBinlogPos;
                            log.info("配置的binlog文件和上次同步的binlog文件都已变化，使用当前的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                        }
                    } else {
                        // 如果没有上次同步的位置，使用当前的binlog位置
                        useBinlogFile = currentBinlogFile;
                        useBinlogPos = currentBinlogPos;
                        log.info("配置的binlog文件已变化且没有上次同步记录，使用当前的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                    }
                }
            } else {
                // 如果配置中没有指定binlog文件和位置，检查是否有上次同步的位置
                if (StringUtils.hasText(lastBinlogFile) && lastBinlogPos > 0) {
                    // 如果有上次同步的位置，检查上次同步的binlog文件是否与当前的相同
                    if (lastBinlogFile.equals(currentBinlogFile)) {
                        // 如果相同，使用上次同步的position
                        useBinlogFile = lastBinlogFile;
                        useBinlogPos = lastBinlogPos;
                        log.info("使用上次同步的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                    } else {
                        // 如果不同，说明binlog文件已经轮换，使用当前的binlog位置
                        useBinlogFile = currentBinlogFile;
                        useBinlogPos = currentBinlogPos;
                        log.info("上次同步的binlog文件已变化，使用当前的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                    }
                } else {
                    // 如果没有上次同步的位置，使用当前的binlog位置
                    useBinlogFile = currentBinlogFile;
                    useBinlogPos = currentBinlogPos;
                    log.info("没有配置和上次同步记录，使用当前的binlog位置: {}:{}", useBinlogFile, useBinlogPos);
                }
            }

            // 设置binlog文件名和位置
            binlogClient.setBinlogFilename(useBinlogFile);
            binlogClient.setBinlogPosition(useBinlogPos);

            // 注册事件监听器
            binlogClient.registerEventListener(event -> {
                try {
                    // 处理binlog事件
                    processDataEvent(event);

                    // 记录当前的binlog位置
                    String currentBinlogFilename = binlogClient.getBinlogFilename();
                    long currentBinlogPosition = binlogClient.getBinlogPosition();

                    // 更新上次同步的binlog位置
                    syncConfig.setLastBinlogFilename(currentBinlogFilename);
                    syncConfig.setLastBinlogPosition(currentBinlogPosition);

                    // 每100个事件更新一次配置文件中的位置
                    if (event.getHeader().getEventType() == EventType.XID) {
                        updateBinlogPosition(currentBinlogFilename, currentBinlogPosition);
                    }
                } catch (Exception e) {
                    log.error("处理binlog事件失败: {}", e.getMessage(), e);
                }
            });

            // 启动binlog客户端
            isRunning = true;
            log.info("增量同步已启动");
            new Thread(() -> {
                try {
                    binlogClient.connect();
                } catch (Exception e) {
                    log.error("Binlog客户端连接失败: {}", e.getMessage(), e);
                    isRunning = false;
                }
            }).start();
        } catch (Exception e) {
            log.error("启动增量同步失败: {}", e.getMessage(), e);
            isRunning = false;
        }
    }

    /**
     * 停止增量同步
     */
    @Override
    public void stopIncrementalSync() {
        if (!isRunning || binlogClient == null) {
            return;
        }
        
        log.info("停止增量同步");
        
        try {
            binlogClient.disconnect();
        } catch (IOException e) {
            log.error("Binlog客户端断开连接失败: {}", e.getMessage(), e);
        } finally {
            isRunning = false;
            binlogClient = null;
        }
    }

    /**
     * 获取表结构
     * 
     * @param tableName 表名
     * @return 表结构信息
     */
    @Override
    public TableSchema getTableSchema(String tableName) {
        log.info("获取表结构: {}", tableName);
        
        // 从缓存获取
        TableSchema cachedSchema = tableSchemaCache.get(tableName);
        if (cachedSchema != null) {
            return cachedSchema;
        }
        
        // 查询表结构
        TableSchema tableSchema = new TableSchema();
        tableSchema.setTableName(tableName);
        
        // 查询列信息
        List<ColumnInfo> columns = mysqlJdbcTemplate.query(
            "SELECT COLUMN_NAME, DATA_TYPE, COLUMN_DEFAULT, IS_NULLABLE, COLUMN_KEY, COLUMN_COMMENT " +
            "FROM INFORMATION_SCHEMA.COLUMNS " +
            "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
            "ORDER BY ORDINAL_POSITION",
            (rs, rowNum) -> {
                ColumnInfo columnInfo = new ColumnInfo();
                columnInfo.setName(rs.getString("COLUMN_NAME"));
                columnInfo.setMysqlType(rs.getString("DATA_TYPE"));
                columnInfo.setClickhouseType(mapMySqlTypeToClickHouseType(rs.getString("DATA_TYPE")));
                columnInfo.setNullable("YES".equals(rs.getString("IS_NULLABLE")));
                columnInfo.setDefaultValue(rs.getString("COLUMN_DEFAULT"));
                columnInfo.setPrimaryKey("PRI".equals(rs.getString("COLUMN_KEY")));
                columnInfo.setComment(rs.getString("COLUMN_COMMENT"));
                return columnInfo;
            },
            syncConfig.getMysqlDatabase(), tableName
        );
        tableSchema.setColumns(columns);
        
        // 查询主键
        List<String> primaryKeys = columns.stream()
            .filter(ColumnInfo::isPrimaryKey)
            .map(ColumnInfo::getName)
            .collect(Collectors.toList());
        tableSchema.setPrimaryKeys(primaryKeys);
        
        // 缓存表结构
        tableSchemaCache.put(tableName, tableSchema);
        
        return tableSchema;
    }

    /**
     * 创建ClickHouse表
     * 
     * @param tableSchema 表结构
     * @return 是否创建成功
     */
    @Override
    public boolean createClickHouseTable(TableSchema tableSchema) {
        log.info("创建ClickHouse表: {}", tableSchema.getTableName());

        // 如果配置了删除表的选项，则先删除已有的表
        if (syncConfig.isSyncDeleteExistingTables()) {
            String dropSql = String.format("DROP TABLE IF EXISTS %s", tableSchema.getTableName());
            try {
                clickhouseJdbcTemplate.execute(dropSql);
                log.info("已删除表: {}", tableSchema.getTableName());
            } catch (Exception e) {
                log.error("删除表失败: {}", e.getMessage());
                return false;
            }
        }

        // 构建建表SQL
        StringBuilder sql = new StringBuilder();
        sql.append("CREATE TABLE IF NOT EXISTS ")
            .append(tableSchema.getTableName())
            .append(" (\n");
        
        // 添加字段定义
        for (int i = 0; i < tableSchema.getColumns().size(); i++) {
            ColumnInfo column = tableSchema.getColumns().get(i);
            sql.append("    ")
                .append(column.getName())
                .append(" ")
                .append(column.getClickhouseType());
            
            if (!column.isNullable()) {
                sql.append(" NOT NULL");
            }
            
            // 添加字段注释
            if (StringUtils.hasText(column.getComment())) {
                sql.append(" COMMENT '").append(column.getComment().replace("'", "\\'")).append("'");
            }
            
            sql.append(",\n");
        }

        // 添加 version 和 is_deleted 字段
        sql.append("    version UInt64 COMMENT '记录版本号（毫秒级时间戳）',\n"); // 添加注释
        sql.append("    is_deleted UInt8 DEFAULT 0 COMMENT '是否删除标记（0=未删除，1=已删除）'\n"); // 添加注释
        
        // 添加主键定义
        if (!tableSchema.getPrimaryKeys().isEmpty()) {
            sql.append(") ENGINE = MergeTree()\n")
                .append("ORDER BY (")
                .append(String.join(", ", tableSchema.getPrimaryKeys()))
                .append(")\n");
        } else {
            sql.append(") ENGINE = MergeTree()\n")
                .append("ORDER BY tuple()\n");
        }
        
        // 执行建表SQL
        try {
            clickhouseJdbcTemplate.execute(sql.toString());
            return true;
        } catch (Exception e) {
            log.error("创建ClickHouse表失败: {}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * 获取同步状态
     * 
     * @return 同步状态信息
     */
    @Override
    public String getSyncStatus() {
        return isRunning ? "增量同步运行中" : "未运行同步";
    }

    /**
     * 处理binlog数据事件
     * 
     * @param event binlog事件
     */
    private void processDataEvent(Event event) {
        EventData data = event.getData();

        if (data instanceof TableMapEventData) {
            TableMapEventData tableData = (TableMapEventData) data;
            tableIdToNameMap.put(tableData.getTableId(), tableData.getTable());
        } else if (data instanceof WriteRowsEventData) {
            // 插入事件
            WriteRowsEventData writeData = (WriteRowsEventData) data;
            String tableName = getTableName(writeData.getTableId());
            
            if (tableName == null) {
                return;
            }
            
            TableSchema tableSchema = tableSchemaCache.get(tableName);
            List<DataRow> dataRows = new ArrayList<>();
            
            for (Object[] row : writeData.getRows()) {
                DataRow dataRow = convertToDataRow(tableSchema, row);
                dataRow.setTableName(tableName);
                
                // 设置 version 为当前时间戳，isDeleted 为 0（新增）
                dataRow.setVersion(System.currentTimeMillis());
                dataRow.setIsDeleted((byte) 0);
                
                dataRows.add(dataRow);
            }
            
            batchInsertToClickHouse(tableSchema, dataRows);
        } else if (data instanceof UpdateRowsEventData) {
            // 更新事件
            UpdateRowsEventData updateData = (UpdateRowsEventData) data;
            long tableId = updateData.getTableId(); // 获取表 ID
            String tableName = tableIdToNameMap.get(tableId); // 根据表 ID 获取表名

            if (tableName == null) {
                log.warn("未找到表 ID {} 对应的表名", tableId);
                return; // 如果找不到表名，直接返回
            }

            TableSchema tableSchema = tableSchemaCache.get(tableName);
            List<DataRow> dataRows = new ArrayList<>();
            for (Map.Entry<Serializable[], Serializable[]> entry : updateData.getRows()) {
                // 以新值为准
                Serializable[] newRow = entry.getValue();
                DataRow dataRow = convertToDataRow(tableSchema, newRow);
                dataRow.setTableName(tableName);
                
                // 设置 version 为当前时间戳，isDeleted 为 0（更新）
                dataRow.setVersion(System.currentTimeMillis());
                dataRow.setIsDeleted((byte) 0);
                
                dataRows.add(dataRow);
            }
            
            batchInsertToClickHouse(tableSchema, dataRows);
        } else if (data instanceof DeleteRowsEventData) {
            // 删除事件 - 通过逻辑删除处理
            DeleteRowsEventData deleteData = (DeleteRowsEventData) data;
            long tableId = deleteData.getTableId();
            String tableName = tableIdToNameMap.get(tableId);
            
            if (tableName == null) {
                log.warn("未找到表 ID {} 对应的表名", tableId);
                return;
            }
            
            TableSchema tableSchema = tableSchemaCache.get(tableName);
            List<DataRow> dataRows = new ArrayList<>();
            
            for (Object[] row : deleteData.getRows()) {
                DataRow dataRow = convertToDataRow(tableSchema, row);
                dataRow.setTableName(tableName);
                
                // 设置 version 为当前时间戳，isDeleted 为 1（删除）
                dataRow.setVersion(System.currentTimeMillis());
                dataRow.setIsDeleted((byte) 1);
                
                dataRows.add(dataRow);
            }
            
            batchInsertToClickHouse(tableSchema, dataRows);
            log.debug("处理删除事件，将数据标记为已删除");
        }
    }

    /**
     * 根据表ID获取表名
     * 
     * @param tableId 表ID
     * @return 表名
     */
    private String getTableName(long tableId) {
        return tableIdToNameMap.get(tableId);
    }

    /**
     * 将原始数据行转换为DataRow对象
     * 
     * @param tableSchema 表结构
     * @param row 原始数据行
     * @return DataRow对象
     */
    private DataRow convertToDataRow(TableSchema tableSchema, Object[] row) {
        DataRow dataRow = new DataRow();
        
        List<ColumnInfo> columns = tableSchema.getColumns();
        for (int i = 0; i < columns.size() && i < row.length; i++) {
            ColumnInfo columnInfo = columns.get(i);
            Object value = row[i];
            dataRow.setValue(columnInfo.getName(), value);
        }
        
        // 提取主键值
        Map<String, Object> primaryKeyValues = new HashMap<>();
        for (String primaryKey : tableSchema.getPrimaryKeys()) {
            primaryKeyValues.put(primaryKey, dataRow.getValue(primaryKey));
        }
        
        // 设置默认值，这些将在各事件处理中被重新设置
        dataRow.setVersion(System.currentTimeMillis());
        dataRow.setIsDeleted((byte) 0);
        
        return dataRow;
    }

    /**
     * 批量插入数据到ClickHouse
     * 
     * @param tableSchema 表结构
     * @param dataRows 数据行列表
     */
    private void batchInsertToClickHouse(TableSchema tableSchema, List<DataRow> dataRows) {
        if (dataRows.isEmpty()) {
            return;
        }
        
        String tableName = tableSchema.getTableName();
        List<ColumnInfo> columns = tableSchema.getColumns();
        
        // 构建插入SQL
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO ").append(tableName).append(" (");
        
        // 添加字段名
        for (int i = 0; i < columns.size(); i++) {
            sql.append(columns.get(i).getName());
            if (i < columns.size() - 1) {
                sql.append(", ");
            }
        }
        
        sql.append(", version, is_deleted) VALUES "); // 添加 version 和 is_deleted 字段
        
        // 添加参数占位符
        for (int i = 0; i < dataRows.size(); i++) {
            sql.append("(");
            for (int j = 0; j < columns.size(); j++) {
                sql.append("?");
                if (j < columns.size() - 1) {
                    sql.append(", ");
                }
            }
            sql.append(", ?, ?)"); // 添加 version 和 is_deleted 的占位符
            
            if (i < dataRows.size() - 1) {
                sql.append(", ");
            }
        }
        
        // 准备参数
        List<Object> params = new ArrayList<>();
        for (DataRow dataRow : dataRows) {
            for (ColumnInfo column : columns) {
                Object value = dataRow.getValue(column.getName());
                if (value instanceof byte[]) {
                    value = new String((byte[]) value, StandardCharsets.UTF_8);
                }
                params.add(value);
            }
            params.add(dataRow.getVersion()); // 添加 version 的值
            params.add(dataRow.getIsDeleted()); // 添加 is_deleted 的值
        }
        
        // 执行插入
        try {
            clickhouseJdbcTemplate.update(sql.toString(), params.toArray());
            log.debug("成功插入 {} 条数据到 {}", dataRows.size(), tableName);
        } catch (Exception e) {
            log.error("插入数据到ClickHouse失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 将MySQL数据类型映射为ClickHouse数据类型
     * 
     * @param mysqlType MySQL数据类型
     * @return ClickHouse数据类型
     */
    private String mapMySqlTypeToClickHouseType(String mysqlType) {
        switch (mysqlType.toLowerCase()) {
            case "tinyint":
                return "Int8";
            case "smallint":
                return "Int16";
            case "int":
            case "integer":
                return "Int32";
            case "bigint":
                return "Int64";
            case "float":
                return "Float32";
            case "double":
                return "Float64";
            case "decimal":
                return "Decimal(38, 10)";
            case "char":
            case "varchar":
            case "text":
                return "String";
            case "date":
                return "Date";
            case "datetime":
            case "timestamp":
                return "DateTime";
            case "boolean":
            case "bool":
                return "UInt8";
            default:
                return "String";
        }
    }

    private String[] getBinlogStatus() {
        String sql = "SHOW MASTER STATUS";
        List<String[]> result = mysqlJdbcTemplate.query(sql, (rs, rowNum) -> new String[]{
            rs.getString("File"),
            rs.getString("Position")
        });
        return result.isEmpty() ? null : result.get(0);
    }

    /**
     * 更新binlog位置到配置文件
     * @param filename binlog文件名
     * @param position binlog位置
     */
    private void updateBinlogPosition(String filename, long position) {
        try {
            // 调用API更新binlog位置
            Map<String, Object> requestBody = new HashMap<>();
            requestBody.put("filename", filename);
            requestBody.put("position", position);
            
            // 这里可以使用RestTemplate或其他HTTP客户端调用API
            // 为了简单起见，这里直接更新内存中的配置
            syncConfig.setLastBinlogFilename(filename);
            syncConfig.setLastBinlogPosition(position);
            
            log.debug("已更新binlog位置: {}:{}", filename, position);
        } catch (Exception e) {
            log.error("更新binlog位置失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 初始化表ID到表名的映射
     * @param host MySQL主机
     * @param port MySQL端口
     * @param username MySQL用户名
     * @param password MySQL密码
     * @param database MySQL数据库名
     */
    private void initTableIdMap(String host, int port, String username, String password, String database) {
        String url = "jdbc:mysql://" + host + ":" + port + "/" + database;
        try (Connection conn = DriverManager.getConnection(url, username, password);
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery("SHOW TABLES")) {
            
            // 清空映射
            tableIdToNameMap.clear();
            
            // 获取所有表名
            List<String> tableNames = new ArrayList<>();
            while (rs.next()) {
                tableNames.add(rs.getString(1));
            }
            
            log.info("初始化表ID映射，共找到 {} 个表", tableNames.size());
            
            // 这里我们无法直接获取表ID，但在处理TableMapEventData事件时会自动建立映射
        } catch (SQLException e) {
            log.error("初始化表ID映射失败: {}", e.getMessage(), e);
        }
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        if (syncConfig.getLastBinlogPosition()<=0&& syncConfig.isFullSync()) {
            init();
        }
        if (syncConfig.isIncrementalSync() && syncConfig.isBinlogEnabled()) {
            startIncrementalSync();
        }
    }
}